package com.example.kafka.consumer;

import com.example.kafka.service.IndicatorDataService;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
import org.springframework.util.StopWatch;

import java.util.Optional;

@Component
@Slf4j
public class KafkaConsumer {

    @Autowired
    IndicatorDataService indicatorDataService;

    @KafkaListener(topics = {"test6"})
    public void consumer(ConsumerRecord<?,?> consumerRecord){
        StopWatch stopWatch = new StopWatch();
        stopWatch.start();
        //判断是否为null
        Optional<?> kafkaMessage = Optional.ofNullable(consumerRecord.value());
        Optional<?> kafkakeyMessage = Optional.ofNullable(consumerRecord.key());
        log.info(">>>>>>>>>> record =" + kafkaMessage);
        log.info(">>>>>>>>>> record =" + kafkakeyMessage);
        if(kafkaMessage.isPresent()){
            //得到Optional实例中的值
            Object message = kafkaMessage.get();
            System.err.println("消费消息:"+message);
        }
        stopWatch.stop();
        log.info("send mock metric data to kafka, cost {} second",  stopWatch.getTotalTimeSeconds());

    }

}
